package com.markhsiu.MiniMQ.remote.transport.netty;

import com.markhsiu.MiniMQ.broker.factory.BrokerFactory;
import com.markhsiu.MiniMQ.channel.ChannelProcessor;
import com.markhsiu.MiniMQ.model.Message;
import com.markhsiu.MiniMQ.model.ResponseMessage;
import com.markhsiu.MiniMQ.model.constant.MessageSourceEnum;
import com.markhsiu.MiniMQ.remote.AbstractMessageProcessor;

/**
 * 
 * @author Mark Hsiu
 *
 */
public class NettyCustomerMessageProcessor extends AbstractMessageProcessor{

	 
	@Override
	public void handler(final Message request,ChannelProcessor channel) {
		if(checkMassage(request)){
			channel.writeAndFlush(ResponseMessage.newError());
			return;
		}
		
		Message response = Message.newSuccess();
		response.setMessageID(request.getMessageID());
		response.setSource(MessageSourceEnum.BROKER);
		response.setTarget(MessageSourceEnum.CUSTOMER);
		
		String consumerID = request.getConsumerID();
		String topic = request.getTopic();
		boolean flag =  true;
		BrokerFactory broker = BrokerFactory.instance();
		switch (request.getCmd()) {
			case CONNECT:
				flag = broker.addConsumer(consumerID);
				broker.addConsumerConnection(consumerID, channel);
				break;
			case DISCONNECT:
				flag = broker.removeConsumer(consumerID);
				broker.removeConsumer(consumerID);
				break;
			case SUB:
				flag = broker.addConsumerSubscribeTopic(consumerID, topic);
				//启动任务
				break;
			case UNSUB:
				flag = broker.removeConsumerSubscribeTopic(consumerID, topic);
				break;
			case SUCCESS:
				return;
			case FAIL:
				System.out.println("消息推送失败！");
				return;
			default:
				response = Message.newFail();
				break;
		} 
		if(!flag){
			response = Message.newFail();
		}
		channel.writeAndFlush(response);
	}

}
